import java.net.DatagramPacket;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.Random;

/**
 * 
 * This class allows setting application class specific policies during 
 * initialization.
 * It also maintains application class specific queues
 * and provides methods used by sender thread(MDRRSender), 
 * receiver thread (MDRRReceiver) and credits manager(CreditsManager)
 * 
 *
 */
public class MDRRBuffer {
	public static final int MAX_PACKET_SIZE = 1490;
	private static final boolean USE_PLAIN_DRR = false;

	private Map<Integer, Float> policies;
	
	//map of port number to BlockingQueue
	Map<Integer, BlockingQueue<DatagramPacket>> queueMap;
	BlockingQueue<DatagramPacket> defaultQueue;
	private int dropCount;
	
	Map<Integer, Integer> creditsMap;
	int defaultCredits;
	
	private List<Integer> eligiblePorts;
	private Map<Integer, Integer> randRanges = new HashMap<Integer, Integer>();
	private Random randGenerator = new Random();
	
	/**
	 * 
	 * @param policies (Map of port number to percentage of bandwidth allocated) 
	 * @param maxQueueLength
	 */
	public MDRRBuffer(Map<Integer, Float> policies, int maxQueueLength)
	{
		this.dropCount = 0;
		this.policies = policies;
		queueMap = new HashMap<Integer, BlockingQueue<DatagramPacket>>();
		creditsMap = new HashMap<Integer, Integer>();
		for(int port: policies.keySet())
		{
			queueMap.put(port, new LinkedBlockingQueue<DatagramPacket>(maxQueueLength));
			creditsMap.put(port, MAX_PACKET_SIZE);
		}
		
		defaultQueue = new LinkedBlockingQueue<DatagramPacket>(maxQueueLength);
		defaultCredits = MAX_PACKET_SIZE;
		
		eligiblePorts = new ArrayList<Integer>();
		initRandRanges();
	}
	
	private void initRandRanges()
	{
		int start = 0;
		for(int port : policies.keySet())
		{
			int value = start + (int) (policies.get(port) * 100);
			randRanges.put(port, value);
			start = start + value;
		}
	}
	
	private synchronized void incrementDropCount()
	{
		dropCount++;
	}

	private int getReceiverPort(DatagramPacket packet)
	{
		byte[] data = packet.getData();
        ByteBuffer byteBuffer = ByteBuffer.wrap(data);
        int seq = byteBuffer.getInt();
        int receiverPort = byteBuffer.getInt(); 
        return receiverPort;
	}
	
	//called by receiver
	public synchronized void addPacket(DatagramPacket packet) {
		int port = getReceiverPort(packet);
		if(queueMap.containsKey(port))
		{
			//this packet was for an application class we recognize
			BlockingQueue<DatagramPacket> queue = queueMap.get(port);
			if(!queue.offer(packet))
			{
				incrementDropCount();
			}
			else
			{
				this.notify();
			}
		}
		else
		{
			//this packet was not for any application class we recognize
			//so we add to the default queue
			if(!defaultQueue.offer(packet))
			{
				incrementDropCount();
			}
			else
			{
				this.notify();
			}
		}
	}
	
	//called by the CreditsManager
	public synchronized void startRound()
	{
		for(int port : creditsMap.keySet())
		{
			int existingCredit = creditsMap.get(port);
			int newCredit = existingCredit + MAX_PACKET_SIZE;
			BlockingQueue<DatagramPacket> queue = queueMap.get(port);
			if(queue.size() == 0)
			{
				newCredit = 0; //take away credits if this queue doesn't have any packets to send
			}
			creditsMap.put(port, newCredit);
		}
		if(defaultQueue.size() == 0)
		{
			defaultCredits = 0;
		}
		else
		{
			defaultCredits += MAX_PACKET_SIZE;
		}
	}
	
	//called by the sender
	public boolean hasPackets()
	{
		for(int port: queueMap.keySet())
		{
			BlockingQueue<DatagramPacket> queue = queueMap.get(port);
			if(queue.size() > 0)
			{
				return true;
			}
		}
		if(defaultQueue.size() > 0)
		{
			return true;
		}
		
		return false;
	}
	
	//called by the sender. 
	public synchronized DatagramPacket removePacket()
	{
		eligiblePorts.clear();
		for(int port : creditsMap.keySet())
		{
			int credit = creditsMap.get(port);
			BlockingQueue<DatagramPacket> queue = queueMap.get(port);
			
			//examine head of queue without removing it to check
			//if the packet is eligible for being sent i.e. queue has
			//enough credits
			DatagramPacket packet = queue.peek();
			if(packet != null && packet.getLength() <= credit)
			{
				//this queue is eligible since it has enough credits and 
				//has a packet to send
				eligiblePorts.add(port);
			}
		}
		
		if(eligiblePorts.size() > 0)
		{
			int winningPort = (USE_PLAIN_DRR ? eligiblePorts.get(0) 
					: findWinningPort());
			int credit = creditsMap.get(winningPort);
			BlockingQueue<DatagramPacket> queue = queueMap.get(winningPort);

			//remove head of the queue since this packet will be sent
			DatagramPacket packet = queue.remove();
			credit = credit - packet.getLength(); //spend the required credit to send this packet
			creditsMap.put(winningPort, credit);
			return packet;
		}
		
		//we were not able to find any packet to send in the application specific queues
		//we will now examine the default queue
		DatagramPacket packet = defaultQueue.peek();
		if(packet != null && packet.getLength() <= defaultCredits)
		{
			defaultCredits = defaultCredits - packet.getLength();
			defaultQueue.remove();
			return packet;
		}
		
		//unable to find any packet to send
		return null;
	}

	//select winningPort among all eligible ports using 
	//probability based on policies
	private Integer findWinningPort() {
		int rand = randGenerator.nextInt(100);
		for(int port : eligiblePorts)
		{
			int start = randRanges.get(port);
			int end = start + (int)(policies.get(port) * 100);
			if(rand >=  start && rand <= end)
			{
				return port;
			}
		}
		
		return eligiblePorts.get(0);
	}
}
